package net.sf.calibur.engine;

import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import net.sf.calibur.engine.task.Handler;
import net.sf.calibur.engine.task.TaskInstance;

public class Executor implements Runnable {
	private final static long SHUTDOWN_TIMEOUT = 10000L;

	enum Status {
		stopped, running, shutdown
	};

	/** 接收消息的阻塞队列 */
	private LinkedBlockingQueue<TaskInstance> queue;
	/** 执行处理的线程 */
	private Thread thread;
	/** log对象 */
	private Logger logger;

	/** 自己所属的池 */
	private ExecutorPool parent;
	/** 自己在所属池中的顺序号 */
	private int index;

	/** 当前忙/闲状态 */
	private boolean busy;
	/** 是否正在运行 */
	private Status status = Status.stopped;

	public Executor(ExecutorPool parent, int idx) {
		this.parent = parent;
		this.index = idx;

		logger = LoggerFactory.getLogger(getClass());
		queue = new LinkedBlockingQueue<TaskInstance>();
		thread = new Thread(this, getName());

		busy = false;
	}

	public boolean isBusy() {
		return busy;
	}

	public String getName() {
		return new StringBuffer(parent.getName()).append('-').append(index)
				.toString();
	}

	public void addTask(TaskInstance taskInstance) {
		if (status != Status.running) {
			throw new RuntimeException(
					"Cannot add task to not running executor " + getName()
							+ " status=" + status.name());
		} else {
			queue.add(taskInstance);
		}
	}

	public int getWaitingEventNum() {
		return queue.size();
	}

	public Status getStatus() {
		return status;
	}

	public void start() {
		if (status == Status.stopped) {
			thread.start();
		} else {
			throw new IllegalStateException("Executor " + getName() + " in "
					+ status + ", cannot start it.");
		}
	}

	public void shutdown() {
		if (status != Status.running) {
			return;
		}

		try {
			logger.info("Waiting executor " + getName() + " shutdown ...");
			addTask(new TaskInstance(null, new ShutdownEvent()));
			status = Status.shutdown;

			// 等待执行线程中止
			thread.join(SHUTDOWN_TIMEOUT);
		} catch (Exception e) {
			logger.error("Executor " + getName() + " shutdown be interrupted.",
					e);
		}

		if (thread.isAlive()) { // 执行线程依然活动，强制打断。
			logger.info("Executor " + getName()
					+ " shutdown failed, forcibly interrupt it.");
			thread.interrupt();
		}
	}

	public void run() {
		logger.info("Executor " + getName() + " started.");
		TaskInstance taskInstance = null;
		Handler handler = null;
		status = Status.running;
		

		while (true) {
			handler = null;
			taskInstance = null;
			try {
				try {
					if (status == Status.running) {
						taskInstance = queue.take();
					} else if (status == Status.shutdown) {
						if (queue.size() > 0) {
							taskInstance = queue.take();
						} else {
							break;
						}
					}
				} catch (InterruptedException ie) {
					break;
				}

				if (taskInstance.getEvent() instanceof ShutdownEvent) {
					break;
				} else {
					handler = taskInstance.getTask().getParent();
					handler.prepareExecute(taskInstance);
					handler.executeTask(taskInstance);
					handler.postExecute(taskInstance);
				}
			} catch (Throwable t) {
				logger.error("Executor " + getName()
						+ " execute task fail. task:" + taskInstance, t);
			} finally {
				busy = false;
			}
		}

		status = Status.stopped;
		logger.info("Executor " + getName() + " stopped.");
	}

	@Override
	public String toString() {
		return new StringBuffer().append(getName()).append(',')
				.append(queue.size()).append(',').append(busy).append(',')
				.append(status).toString();
	}
}
